package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo02SourceAPI {
  def main(args: Array[String]): Unit = {
    // 构建SparkSession环境
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local")
      .getOrCreate()

    // 读普通文本文件直接用csv格式
    val csvStuDF: DataFrame = spark
      .read
      .format("csv") // 设置数据的加载格式
      .option("sep", ",") // 设置参数，csv格式加载时分隔符
      .schema("id String,name String,age Int,gender String,clazz String") // 由于csv格式的数据没有自带结构，就需要手动通过schema指定
      .load("spark/data/source/csv")

    // 对多次使用的DF也可以进行缓存
    csvStuDF.cache()

    // show是DF特有的打印数据的方式，每次show都会触发一次job，show类似RDD中的行为算子
    csvStuDF.show() // 默认打印20条数据
    csvStuDF.show(10) // 指定条数打印
    csvStuDF.show(numRows = 10,truncate = false) // truncate参数表示是否截断数据输出，默认为true
    csvStuDF.show(numRows = 10,truncate = 0,vertical = true) // vertical 垂直输出，默认为false

    // 写文件
    csvStuDF
      .write
      /**
      *指定写入模式：
      * append：追加
      * overwrite：覆盖
      * ignore：存在即忽略
      * errorIfExists：存在即报错
      */
      .mode(SaveMode.Overwrite)
      .format("json")
      .save("spark/data/source/json/") // 输出路径只能指定目录，因为ReduceTask会有N个


    csvStuDF.write.mode(SaveMode.Overwrite).format("parquet").save("spark/data/source/parquet/")
    csvStuDF.write.mode(SaveMode.Overwrite).format("orc").save("spark/data/source/orc/")


    // 释放缓存
    csvStuDF.unpersist()

    val jsonStuDF: DataFrame = spark.read.format("json").load("spark/data/source/json")
    val parquetStuDF: DataFrame = spark.read.format("parquet").load("spark/data/source/parquet")
    val orcStuDF: DataFrame = spark.read.format("orc").load("spark/data/source/orc")

    jsonStuDF.show() // json格式每条数据自带结构，空间占用最多，不需要指定schema，最终输出时会按照列名的排序进行输出
    // parquet、orc两者都是列式存储格式，空间占用上都会对数据进行压缩，orc占用的空间最小
    // parquet适合存储嵌套数据，orc对于SQL查询性能提升最高
    parquetStuDF.show()
    orcStuDF.show()

    // JDBC Source
    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/stu/students.txt")

    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,subject_id String,score Int")
      .load("spark/data/stu/score.txt")

    // 计算clazz Top3

    // 第一种方式：DSL
    // 计算总分
    val sumScoreDF: DataFrame = scoreDF
      .groupBy("id")
      .sum("score")
      .withColumnRenamed("sum(score)","sum_score")

    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 关联
    sumScoreDF
      .join(stuDF,"id")
      .select(
        $"id",$"name",$"age",$"clazz",$"sum_score"
        ,row_number() over Window.partitionBy("clazz").orderBy($"sum_score".desc) as "rn"
      )
      .where("rn<=3")
      .write
      .mode(SaveMode.Overwrite)
      .format("jdbc")
      .option("url", "jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32")
      .option("dbtable", "bigdata32.clazz_top3_zzk")
      .option("user", "bigdata32")
      .option("password", "123456")
//      .save()

    // 第二种方式：基于SQL
    // 将准备好的数据先注册成表才能写SQL
//    stuDF.createOrReplaceTempView("stu")
    stuDF.createOrReplaceGlobalTempView("stu") // 会将表默认放在global_temp库下，使用时需要加上库名

    scoreDF.createOrReplaceTempView("score") // 将DF直接注册成临时视图，可以直接使用表名，不需要指定库名

    val clazzTopDF: DataFrame = spark.sql(
      """
        |select *
        |from (
        |select  t1.id
        |        ,t2.name
        |        ,t1.sum_score
        |        ,t2.clazz
        |        ,row_number() over (partition by t2.clazz order by t1.sum_score desc) as rn
        |from (
        |    select id
        |           ,sum(score) as sum_score
        |    from score
        |    group by id
        |) t1 join global_temp.stu t2 on t1.id = t2.id
        |) tt1 where tt1.rn <= 3
        |
        |""".stripMargin)

    clazzTopDF
      .write
      .mode(SaveMode.Overwrite)
      .format("jdbc")
      .option("url", "jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32")
      .option("dbtable", "bigdata32.clazz_top3_zzk_sql")
      .option("user", "bigdata32")
      .option("password", "123456")
//      .save()

    // 读MySQL的数据
    val stuSQLDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32")
      .option("dbtable", "bigdata32.clazz_top3_zzk_sql")
      .option("user", "bigdata32")
      .option("password", "123456")
      .load()

//    stuSQLDF.show(1)

  }

}
